

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.Scanner;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import java.util.function.Consumer;

public class WhatEver {

    public static void main(String[] args) throws Exception {
        ZooKeeper zkCli;
        AtomicInteger count = new AtomicInteger(0);
        Stat stat = new Stat();
        ReentrantLock stoplock = new ReentrantLock();
        Condition stoptag = stoplock.newCondition();
        Consumer<Object> debug = o -> {
            count.getAndIncrement();
            System.out.println(Thread.currentThread().getName() + " : " +
                    count + " : " + o.getClass().getName() + " : " + o);
        };


        zkCli = new ZooKeeper(
                "master:2181",
                3600, (watchedEvent) -> {
            debug.accept("默认回调函数　" + watchedEvent);

        });
        Stat exists = zkCli.exists("/stick", true);
        if (exists == null) {
            System.out.println("/stick 不存在　将会自动创建");
            zkCli.create("/stick", "aotuwin".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        byte[] data = zkCli.getData("/stick", event -> {
            debug.accept("主动监听写回调" + event);
        }, stat);
        debug.accept(new String(data));
        System.out.println("stat ?? " + stat);

////        回调函数x表示类似http请求的get查询　，一次请求一次回复
//            Consumer<byte[]> x  = d ->{
//                debug.accept(new String(d));
//            };
////        回调函数y表示类似ws的订阅　聊天室　，一次订阅多次回复
//            Consumer<WatchedEvent> y  = e ->{
//                debug.accept(e);
//            };
//            BiConsumer<Consumer,Consumer> searchData = (x1,y1) ->{
//                byte[] data1 = zkCli.getData("/stick", (Watcher) y1, null);
//                x1.accept(data1);
//            };
//            searchData(x,y);


        new Thread(() -> {
            Scanner scanner = new Scanner(System.in);
            System.out.println(Thread.currentThread().getName());
            while (true) {
                String s = scanner.nextLine();
                stoplock.lock();
                try {
                    stoptag.signal();
                } finally {
                    stoplock.unlock();
                }
                if (s.equals("q")) break;
                else {
                    try {
                        Stat ss = zkCli.exists("/stick", false);
                        zkCli.setData("/stick",s.getBytes(),ss.getVersion());

                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }, Thread.currentThread().getName() + "-MyThread").start();
        stoplock.lock();
        try {
            stoptag.await();
        } catch (InterruptedException e) {
            System.out.println("程序主线程被意外中断");
        } finally {
            stoplock.unlock();
        }
    }


}
